package sample.persistence

import akka.actor._
import akka.persistence._

class ExamplePersistentActor extends PersistentActor {
  override def persistenceId = "sample-id-2"

  var received: List[String] = Nil // state

  //override def preStart(): Unit = deleteMessages(5)

  def receiveCommand: Receive = {
    case "print" => println(s"received ${received.reverse}")
    case "boom"  => throw new Exception("boom")
    case payload: String =>
      persist(payload) { p => received = p :: received }
    case DeleteMessagesSuccess(seq) =>
      println("seq " + seq + " has deleted ok")
    case DeleteMessagesFailure(cause,seq) =>
      println("seq " + seq + " deleted failed:" + cause)
  }

  def receiveRecover: Receive = {
    case s: String => received = s :: received
  }

  //默认将停止actor,不是重启。因为重启很有可能还是失败
  override protected def onPersistFailure(cause: Throwable, event: Any, seqNr: Long): Unit = super.onPersistFailure(cause,event,seqNr)

  //在存储之前就失败了，比如序列化错误，默认动作是打印warning,继续处理
  override protected def onPersistRejected(cause: Throwable, event: Any, seqNr: Long): Unit = super.onPersistRejected(cause,event,seqNr)

  //恢复失败，默认也将停止actor，不管是从journal恢复还是从snapshot恢复，当然如果得知snapshot已不兼容，可以禁止从它恢复
  override protected def onRecoveryFailure(cause: Throwable, event: Option[Any]): Unit = super.onRecoveryFailure(cause, event)
}

object PersistentActorFailureExample extends App {
  val system = ActorSystem("example")
  val persistentActor = system.actorOf(Props(classOf[ExamplePersistentActor]), "persistentActor-2")

  persistentActor ! "a"
  persistentActor ! "print"
  persistentActor ! "boom" // restart and recovery
  persistentActor ! "print"
  persistentActor ! "b"
  persistentActor ! "print"
  persistentActor ! "c"
  persistentActor ! "print"

  // Will print in a first run (i.e. with empty journal):

  // received List(a)
  // received List(a)
  // received List(a, b)
  // received List(a, b, c)

  // Will print in a second run:

  // received List(a, b, c, a)
  // received List(a, b, c, a)
  // received List(a, b, c, a, b)
  // received List(a, b, c, a, b, c)

  // etc ...

  Thread.sleep(10000)
  system.terminate()
}
